package com.shujia.flink.tf

import org.apache.flink.streaming.api.scala._

object Demo6Agg {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val studentsDS: DataStream[String] = env.readTextFile("flink/data/students.txt")

    /**
      * 聚合函数
      * sum
      * max
      * min
      * maxBy
      * minBy
      *
      */


    /**
      * 获取每隔班级最大的年龄
      *
      */

    val kvDS: DataStream[(String, Int)] = studentsDS.map(line => {
      val split: Array[String] = line.split(",")
      (split(4), 1)
    })

    val maxAge: DataStream[(String, Int)] = kvDS
      .keyBy(_._1)
      .sum(1) //传入下标，下标从0开始

    // maxAge.print()

    /**
      *
      * max 和 maxBy 之间的区别在于 max 返回流中的最大值，但 maxBy 返回具有最大值的键， min 和 minBy 同理。
      */

    val stuDS: DataStream[Student] = studentsDS.map(line => {
      val split: Array[String] = line.split(",")
      Student(split(0), split(1), split(2).toInt, split(3), split(4))
    })

    stuDS
      .keyBy(_.clazz)
      .maxBy("age") //如果数据的类型是样例类可以使用列名
      .print()

    env.execute()

  }

}

case class Student(id: String, name: String, age: Int, gender: String, clazz: String)
